package com.example.monitor

import java.sql.{Connection, PreparedStatement}
import java.util.Properties

import com.alibaba.druid.pool.DruidDataSourceFactory
import com.example.bean.BusInfo
import javax.sql.DataSource
import org.apache.spark.sql.ForeachWriter

/**
 * 将车辆经纬度写入mysql
 *
 * @author lbq
 * @date 2022/3/13 15:20
 * @version 1.0
 */
object MysqlWriter {

  val properties = new Properties()
  properties.setProperty("driverClassName", "com.mysql.jdbc.Driver")
  properties.setProperty("url", "jdbc:mysql:///lagou_study?serverTimezone=Asia/Shanghai&useSSL=false&allowPublicKeyRetrieval=true")
  properties.setProperty("username", "root")
  properties.setProperty("password", "root")
  properties.setProperty("initialSize", "5")
  properties.setProperty("maxActive", "10")
  properties.setProperty("maxWait", "3000")
  properties.setProperty("maxIdle", "6")
  properties.setProperty("minIdle", "3")
  val dataSource: DataSource = DruidDataSourceFactory.createDataSource(properties)

  /**
   * 获取mysql连接
   *
   * @return
   */
  def getConnection(): Connection = {
    dataSource.getConnection
  }
}

class MysqlWriter extends ForeachWriter[BusInfo] {

  var conn: Connection = _

  override def open(partitionId: Long, epochId: Long): Boolean = {
    conn = MysqlWriter.getConnection()
    true
  }

  override def process(value: BusInfo): Unit = {
    val lglat: String = value.lglat
    val arr: Array[String] = lglat.split("_")
    val deployNum: String = value.deployNum
    val sql: String = "insert into t_gps(deploy_num, lng, lat) values (?, ?, ?)"
    val statement: PreparedStatement = conn.prepareStatement(sql)
    statement.setString(1, deployNum)
    statement.setString(2, arr(0))
    statement.setString(3, arr(1))
    statement.executeUpdate()
    statement.close()
  }

  override def close(errorOrNull: Throwable): Unit = {
    conn.close()
  }
}
